#!/bin/env python
# -*- coding: UTF-8 -*-

import urllib2
import json
import os
import time
import logging
from logging import handlers


def listJob(url, applications):
    """
    获取任务列表
    """
    u = urllib2.urlopen(url)
    data = u.read()
    d = json.JSONDecoder()
    jo = d.decode(data)
    apps = jo.pop('apps')
    if apps is not None and 'app' in apps:
        apps = apps.pop('app')
        for item in apps:
            if 'name' in item:
                applications[item.pop('name')] = item.pop('id')
    else:
        pass

def appIsHealth(host, appid):
    """
    检查任务状态
    """
    u = urllib2.urlopen(host + '/proxy/' + appid + '/api/v1/applications/' + appid + '/jobs?status=running')
    data = u.read()
    d = json.JSONDecoder()
    jobs = d.decode(data)
    for job in jobs:
        submissionTime = time.strptime(job.pop('submissionTime'), '%Y-%m-%dT%H:%M:%S.%fGMT')
        numTasks = job.pop('numTasks')
        numFailedTasks = job.pop('numFailedTasks')
        f = numFailedTasks*1.0/numTasks
	    #如果job运行时间超过5分钟且task失败率大于0.2，则认为application故障
        if time.mktime(time.gmtime()) - time.mktime(submissionTime) > 300 and f > 0.2:
	        return False
    return True

def checkApplicationStatus(url, status):
    u = urllib2.urlopen(url)
    data = u.read()
    d = json.JSONDecoder()
    state = d.decode(data).pop('state')
    if state == status:
	return True
    return False	

if __name__ == "__main__":
    filehandler=logging.handlers.TimedRotatingFileHandler("/home/bbp/application-status.log",'D')
    filehandler.setFormatter(logging.Formatter('%(levelname)s %(asctime)s %(message)s'))
    log = logging.getLogger('JobChecker')
    log.addHandler(filehandler)
    log.level = logging.INFO
    log.info('-------------Check Spark Streaming Application---------->')

    apps = {}
    host = 'http://192.168.112.21:8088'
    path = '/home/bbp/'
    #GET SUBMITTED APPS
    listJob(host + '/ws/v1/cluster/apps?states=SUBMITTED', apps)
    #GET ACCEPTED APPS
    listJob(host + '/ws/v1/cluster/apps?states=ACCEPTED', apps)
    #GET RUNIING APPS
    listJob(host + '/ws/v1/cluster/apps?states=RUNNING', apps)

    shouldExistApps = {'com.cetc.sdp.kmga.cs.stream.StateStream':path+'dev.sh', 'com.cetc.sdp.kmga.cs.stream.nv.AuditStreamDriver':path+'audit_driver.sh', 'com.cetc.sdp.kmga.cs.stream.nv.EventStreamDriver':path+'event_driver.sh'}
    for appName in shouldExistApps:
        cmd = shouldExistApps[appName]
        if appName not in apps.keys():
            log.error(appName + ' was stoped.')
            exit = os.system(cmd)
            log.info('Restart ' + appName + ' success.')
        else:
            log.info(appName + ' is active.')
            #检查任务健康状态,
            #http://www.namenode.com:8088/proxy/[application-id]/api/v1/applications/[application-id]/jobs?status=[running|failed|succeed|unknown]
            appId = apps[appName]
            if appIsHealth(host, appId) is not True:
                log.error(appName + ' is not health.')
		#kill application
		os.system('curl ' + host + '/ws/v1/cluster/apps/' + appId + '/state -X PUT -i -H "Content-Type:application/json" -H "Accept:application/json" -d \'{"state":"KILLED"}\'')
		#if application was killed, then restart
		if checkApplicationStatus(host + '/ws/v1/cluster/apps/' + appId + '/state', 'KILLED'):
		    os.system(cmd)
		    log.info('Restart ' + appName + ' success.')
            else:
                log.info(appName + ' is health.')
    log.info('<----------OVER-------------------')
